/* -------------------------------------------------------------------------
 *  This file is part of the Cantian project.
 * Copyright (c) 2023 Huawei Technologies Co.,Ltd.
 *
 * Cantian is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *
 *          http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 * -------------------------------------------------------------------------
 *
 * bak_build.c
 *
 *
 * IDENTIFICATION
 * src/kernel/backup/bak_build.c
 *
 * -------------------------------------------------------------------------
 */

#include "bak_build.h"
#include "cm_file.h"
#include "cm_kmc.h"
#include "knl_backup.h"
#include "bak_restore.h"
#include "knl_context.h"
#include "dtc_database.h"

static void bak_set_build_params(bak_t *bak, cs_pipe_t *pipe, cs_packet_t *send_pack, cs_packet_t *recv_pack,
                                 build_param_ctrl_t *ctrl)
{
    bak_remote_t *remote = &bak->remote;
    bak_attr_t *attr = &bak->record.attr;
    char ipstr[CM_MAX_IP_LEN];

    bak->record.start_time = (uint64)cm_now();
    bak->is_building = GS_TRUE;
    bak->record.log_only = GS_FALSE;
    bak->record.data_only = GS_FALSE;
    bak->record.is_repair = ctrl->is_repair;
    bak->record.is_increment = ctrl->is_increment;
    attr->backup_type = BACKUP_MODE_FULL;
    attr->level = (ctrl->is_increment || ctrl->is_repair) ? GS_TRUE : GS_FALSE;
    attr->base_lsn = ctrl->base_lsn;
    attr->compress = ctrl->compress;
    bak->compress_ctx.compress_level = ctrl->compress_level;
    bak->encrypt_info.encrypt_alg = ENCRYPT_NONE;
    bak->record.device = DEVICE_DISK;
    bak->cumulative = GS_FALSE;
    bak->is_first_link = GS_TRUE;
    bak->need_retry = GS_FALSE;
    bak->need_check = GS_FALSE;
    bak->failed = GS_FALSE;
    bak->proc_count = ctrl->parallelism;
    bak->backup_buf_size = ctrl->buffer_size;

    remote->pipe = pipe;
    remote->send_pack = send_pack;
    remote->recv_pack = recv_pack;

    errno_t ret = strncpy_s(bak->peer_host, GS_HOST_NAME_BUFFER_SIZE, cm_inet_ntop(
        (struct sockaddr *)&pipe->link.tcp.remote.addr, ipstr, CM_MAX_IP_LEN), GS_HOST_NAME_BUFFER_SIZE - 1);
    knl_securec_check(ret);

    GS_LOG_RUN_INF("[BUILD] is increment %u, base lsn %llu ", ctrl->is_increment, ctrl->base_lsn);
}

bool32 knl_brain_repair_check(knl_session_t *session)
{
    reset_log_t *reset_log = &session->kernel->lrcv_ctx.primary_resetlog;
    knl_instance_t *kernel = (knl_instance_t *)session->kernel;
    log_point_t *rcy = &dtc_my_ctrl(session)->rcy_point;

    if ((kernel->lrcv_ctx.dbid != 0) && (kernel->lrcv_ctx.dbid != kernel->db.ctrl.core.dbid)) {
        GS_LOG_RUN_ERR("[REPAIR] primary dbid [%u] is not equal to standby dbid [%u], need repair",
            kernel->lrcv_ctx.dbid, kernel->db.ctrl.core.dbid);
        GS_THROW_ERROR(ERR_INVALID_OPERATION, ", primary dbid is not equal to standby dbid.");
        return GS_FALSE;
    }

    if (reset_log->last_asn > rcy->asn || rcy->rst_id >= reset_log->rst_id) {
        GS_LOG_RUN_ERR("[REPAIR] reset_log and rcy point not matched. reset_log rst_id-asn-lfn: %u-%u-%llu,"
            "rcy point rst_id-asn-lfn: %u-%u-%llu.", reset_log->rst_id, reset_log->last_asn, reset_log->last_lfn,
            rcy->rst_id, rcy->asn, (uint64)rcy->lfn);
        GS_THROW_ERROR(ERR_INVALID_OPERATION, ", asn of reset_log is larger than asn of rcy or rst_id of rcy is larger "
            "than rst_id of reset_log.");
        return GS_FALSE;
    }

    uint32 log_count = log_get_count(session);
    if (rcy->asn - reset_log->last_asn >= log_count) {
        GS_LOG_RUN_ERR("[REPAIR] the count of log between rcy(asn: %u) and reset_log(asn: %u) is larger than "
            "log_count %u.", rcy->asn, reset_log->last_asn, log_count);
        GS_THROW_ERROR(ERR_INVALID_OPERATION, ", log generated by standby is too large.");
        return GS_FALSE;
    }

    if (reset_log->rst_id - rcy->rst_id > 1) {
        GS_LOG_RUN_ERR("[REPAIR] reset_log and rcy point not matched. reset_log rst_id-asn-lfn: %u-%u-%llu,"
            "rcy point rst_id-asn-lfn: %u-%u-%llu.", reset_log->rst_id, reset_log->last_asn, reset_log->last_lfn,
            rcy->rst_id, rcy->asn, (uint64)rcy->lfn);
        GS_THROW_ERROR(ERR_INVALID_OPERATION, ", rst_id of reset_log and rcy is not mapped.");
        return GS_FALSE;
    }

    if (reset_log->last_lfn > rcy->lfn) {
        GS_LOG_RUN_ERR("[REPAIR] primary lfn [%llu] is larger than standby rcy lfn [%llu]", reset_log->last_lfn, (uint64)rcy->lfn);
        GS_THROW_ERROR(ERR_INVALID_OPERATION, ", primary lfn is larger than standby rcy lfn.");
        return GS_FALSE;
    }

    return GS_TRUE;
}

status_t build_analyze_mem_init(knl_session_t *session)
{
    bak_t *bak = &session->kernel->backup_ctx.bak;
    build_analyse_bucket_t *free_list = &bak->build_aly_free_list;
    int64 buf_size = BUILD_ALY_MAX_PAGE_SIZE + BUILD_ALY_MAX_ITEM_SIZE + BUILD_ALY_MAX_BUCKET_SIZE;
    errno_t ret;

    if (cm_aligned_malloc(buf_size, "build log analysis", &bak->build_aly_mem) != GS_SUCCESS) {
        return GS_ERROR;
    }
    bak->build_aly_pages = (page_id_t *)bak->build_aly_mem.aligned_buf;
    bak->build_aly_items = (build_analyse_item_t *)(bak->build_aly_mem.aligned_buf + BUILD_ALY_MAX_PAGE_SIZE);
    bak->build_aly_buckets = (build_analyse_bucket_t *)(bak->build_aly_mem.aligned_buf + BUILD_ALY_MAX_PAGE_SIZE +
                                                        BUILD_ALY_MAX_ITEM_SIZE);

    ret = memset_sp(bak->build_aly_mem.aligned_buf, (size_t)buf_size, 0, (size_t)buf_size);
    knl_securec_check(ret);

    free_list->count = BUILD_ALY_MAX_ITEM;
    free_list->first = &bak->build_aly_items[0];
    for (uint32 i = 0; i < BUILD_ALY_MAX_ITEM - 1; i++) {
        bak->build_aly_items[i].next = &bak->build_aly_items[i + 1];  // init free list
        bak->build_aly_items[i].page_id = &bak->build_aly_pages[i];
    }
    bak->build_aly_items[BUILD_ALY_MAX_ITEM - 1].page_id = &bak->build_aly_pages[BUILD_ALY_MAX_ITEM - 1];
    bak->page_count = 0;

    if (cm_aligned_malloc(GS_MAX_BATCH_SIZE, "build analyze log buffer", &bak->log_buf) != GS_SUCCESS) {
        cm_aligned_free(&bak->build_aly_mem);
        return GS_ERROR;
    }

    return GS_SUCCESS;
}

void build_analyze_mem_free(knl_session_t *session)
{
    bak_t *bak = &session->kernel->backup_ctx.bak;

    cm_aligned_free(&bak->build_aly_mem);
    bak->build_aly_pages = NULL;
    bak->build_aly_items = NULL;
    bak->build_aly_buckets = NULL;
    bak->page_count = 0;
    cm_aligned_free(&bak->log_buf);
}

build_analyse_item_t *build_analyze_pop_free_item(knl_session_t *session)
{
    bak_t *bak = &session->kernel->backup_ctx.bak;
    build_analyse_bucket_t *free_list = &bak->build_aly_free_list;
    build_analyse_item_t *item = NULL;

    if (free_list->first == NULL) {
        return NULL;
    }

    item = free_list->first;
    free_list->first = item->next;
    free_list->count--;
    item->next = NULL;

    return item;
}

static void brain_repair_invalidate_page(bak_t *bak, page_id_t *page_id)
{
    uint32 file_hash = page_id->file % BUILD_ALY_MAX_FILE;
    uint32 page_hash = page_id->page % BUILD_ALY_MAX_BUCKET_PER_FILE;
    build_analyse_bucket_t *bucket = &bak->build_aly_buckets[file_hash * BUILD_ALY_MAX_BUCKET_PER_FILE + page_hash];

    build_analyse_item_t *item = bucket->first;
    while (item != NULL) {
        if (IS_SAME_PAGID(*item->page_id, *page_id)) {
            item->page_id->file = GS_INVALID_FILEID;
            break;
        }
        item = item->next;
    }
}

void brain_repair_filer_page_from_remote(knl_session_t *session, page_head_t *page, uint32 page_count)
{
    bak_t *bak = &session->kernel->backup_ctx.bak;
    page_head_t *page_head = page;
    if (!bak->record.is_repair) {
        return;
    }
    char *page_buff = NULL;
    for (uint32 i = 0; i < page_count; i++) {
        page_id_t *page_id = AS_PAGID_PTR(page_head->id);
        brain_repair_invalidate_page(bak, page_id);
        page_buff = (char*)page_head + DEFAULT_PAGE_SIZE(session);
        page_head = (page_head_t *)page_buff;
    }

    return;
}

status_t build_aly_add_page(knl_session_t *session, page_id_t page_id)
{
    bak_t *bak = &session->kernel->backup_ctx.bak;
    uint32 file_hash = page_id.file % BUILD_ALY_MAX_FILE;
    uint32 page_hash = page_id.page % BUILD_ALY_MAX_BUCKET_PER_FILE;
    build_analyse_bucket_t *bucket = &bak->build_aly_buckets[file_hash * BUILD_ALY_MAX_BUCKET_PER_FILE + page_hash];

    build_analyse_item_t *item = bucket->first;
    while (item != NULL) {
        if (IS_SAME_PAGID(*item->page_id, page_id)) {
            return GS_SUCCESS;
        }
        item = item->next;
    }

    build_analyse_item_t *new_item = build_analyze_pop_free_item(session);
    if (new_item == NULL) {
        GS_LOG_RUN_ERR("[BUILD] pages changed by standby is too much.");
        GS_THROW_ERROR(ERR_INVALID_OPERATION, ", pages changed by standby is too much.");
        return GS_ERROR;
    }

    GS_LOG_DEBUG_INF("[REPAIR] standby find page %u-%u", page_id.file, page_id.page);

    new_item->next = bucket->first;
    *new_item->page_id = page_id;
    bucket->first = new_item;
    bucket->count++;
    bak->page_count++;
    return GS_SUCCESS;
}

static void build_analyze_construct_hash(knl_session_t *session)
{
    bak_t *bak = &session->kernel->backup_ctx.bak;
    build_analyse_item_t *item = NULL;
    build_analyse_bucket_t *bucket = NULL;
    uint32 file_hash, page_hash;

    for (uint32 i = 0; i < bak->page_count; i++) {
        item = &bak->build_aly_items[i];
        file_hash = item->page_id->file % BUILD_ALY_MAX_FILE;
        page_hash = item->page_id->page % BUILD_ALY_MAX_BUCKET_PER_FILE;
        bucket = &bak->build_aly_buckets[file_hash * BUILD_ALY_MAX_BUCKET_PER_FILE + page_hash];

        item->next = bucket->first;
        bucket->first = item;
        bucket->count++;
    }
}

status_t build_find_batch(knl_session_t *session, device_type_t type, int32 *handle, log_file_head_t *head,
                          log_point_t *point, uint32 *data_size, aligned_buf_t *align_buf)
{
    uint64 size, remain_size;
    uint64 buf_size = (uint64)align_buf->buf_size;
    bool32 finished = GS_FALSE;
    log_batch_t *batch = NULL;
    log_batch_tail_t *tail = NULL;
    char *buf = align_buf->aligned_buf;
    uint64 last_lfn = 0;

    int64 file_size = cm_file_size(*handle);
    if (file_size == -1) {
        GS_LOG_RUN_ERR("[BUILD] failed to get log [%u-%u] size ", point->rst_id, point->asn);
        GS_THROW_ERROR(ERR_SEEK_FILE, 0, SEEK_END, errno);
        return GS_ERROR;
    }
    uint64 offset = CM_CALC_ALIGN(sizeof(log_file_head_t), (uint32)head->block_size);
    do {
        size = (uint64)file_size - offset;
        size = size > buf_size ? buf_size : size;
        if (size == 0 || finished) {
            break;
        }

        if (cm_read_device(type, *handle, (int64)offset, buf, (int32)size) != GS_SUCCESS) {
            GS_LOG_RUN_ERR("[BUILD] Failed to read log [%u-%u] file", point->rst_id, point->asn);
            return GS_ERROR;
        }

        batch = (log_batch_t *)buf;
        tail = (log_batch_tail_t *)((char *)batch + batch->size - sizeof(log_batch_tail_t));

        remain_size = size;
        while (remain_size >= sizeof(log_batch_t)) {
            if (!rcy_validate_batch(batch, tail) || batch->head.point.rst_id != head->rst_id ||
                (last_lfn != 0 && batch->head.point.lfn != last_lfn + 1)) {
                finished = GS_TRUE;
                GS_LOG_RUN_INF("[BUILD] log [%u-%u] offset %llu invalid batch size %u "
                               "head [%llu/%u-%u/%llu/%llu] latest_lfn %llu",
                               head->rst_id, head->asn, offset, batch->size, batch->head.magic_num,
                               batch->head.point.rst_id, batch->head.point.asn,
                               (uint64)batch->head.point.lfn, batch->raft_index, last_lfn);

                break;
            }
            last_lfn = batch->head.point.lfn;
            if (batch->head.point.lfn == point->lfn) {
                finished = GS_TRUE;
                if (session->kernel->lrcv_ctx.primary_reset_log_scn == batch->scn ||
                    session->kernel->lrcv_ctx.primary_reset_log_scn == 0) {
                    *data_size = batch->size;
                } else {
                    *data_size = 0;
                    GS_LOG_RUN_ERR("[BUILD] For same lfn [%llu] in resetlog, scn from primary and standby [%llu/%llu] "
                                   "is different, break-brain build is not allowed", (uint64)point->lfn,
                                   session->kernel->lrcv_ctx.primary_reset_log_scn, batch->scn);
                }
                break;
            }
            offset += batch->space_size;
            remain_size -= batch->space_size;
            batch = (log_batch_t *)((char *)batch + batch->space_size);
            if (remain_size < batch->space_size) {
                break;
            }
            tail = (log_batch_tail_t *)((char *)batch + batch->size - sizeof(log_batch_tail_t));
        }
    } while (1);

    if (*data_size != 0) {
        errno_t ret = memmove_s(buf, (uint32)buf_size, batch, batch->size);
        knl_securec_check(ret);
    }
    return GS_SUCCESS;
}

status_t build_load_from_online(knl_session_t *session, uint32 file_id, log_point_t *point, uint32 *data_size,
    aligned_buf_t *align_buf)
{
    log_file_head_t head;
    int32 handle = GS_INVALID_HANDLE;
    log_file_t *file = &session->kernel->redo_ctx.files[file_id];

    if (cm_open_device(file->ctrl->name, file->ctrl->type, knl_redo_io_flag(session), &handle) != GS_SUCCESS) {
        GS_LOG_RUN_ERR("[BUILD] failed to open %s", file->ctrl->name);
        return GS_ERROR;
    }

    uint64 size = CM_CALC_ALIGN(sizeof(log_file_head_t), file->ctrl->block_size);
    log_file_head_t *tmp_head = (log_file_head_t *)align_buf->aligned_buf;
    if (cm_read_device(file->ctrl->type, handle, 0, tmp_head, (int32)size) != GS_SUCCESS) {
        cm_close_device(file->ctrl->type, &handle);
        GS_LOG_RUN_ERR("[BUILD] failed to read %s ", file->ctrl->name);
        return GS_ERROR;
    }

    head.block_size = tmp_head->block_size;
    head.rst_id = tmp_head->rst_id;
    head.asn = tmp_head->asn;
    if (build_find_batch(session, file->ctrl->type, &handle, &head, point, data_size, align_buf) != GS_SUCCESS) {
        cm_close_device(file->ctrl->type, &handle);
        return GS_ERROR;
    }

    cm_close_device(file->ctrl->type, &handle);
    return GS_SUCCESS;
}

status_t build_load_from_arch(knl_session_t *session, log_point_t *point, uint32 *data_size, aligned_buf_t *align_buf)
{
    arch_file_t arch_file = { .name = { 0 }, .handle = GS_INVALID_HANDLE };
    bool32 is_compress = GS_FALSE;
    if (rcy_load_arch(session, (uint32)point->rst_id, point->asn, &arch_file, &is_compress) != GS_SUCCESS)  {
        /* max rst_id <= 2^18, cannot overflow */
        cm_close_device(cm_device_type(arch_file.name), &arch_file.handle);
        rcy_close_file(session);
        return GS_ERROR;
    }

    if (build_find_batch(session, cm_device_type(arch_file.name), &arch_file.handle, &arch_file.head, point, data_size,
                         align_buf) != GS_SUCCESS) {
        cm_close_device(cm_device_type(arch_file.name), &arch_file.handle);
        rcy_close_file(session);
        return GS_ERROR;
    }

    cm_close_device(cm_device_type(arch_file.name), &arch_file.handle);
    return GS_SUCCESS;
}

status_t build_load_batch(knl_session_t *session, log_point_t *point, uint32 *data_size, aligned_buf_t *buf)
{
    status_t status = GS_SUCCESS;
    knl_instance_t *kernel = session->kernel;
    bak_context_t *ctx = &kernel->backup_ctx;
    bak_t *bak = &ctx->bak;

    log_lock_logfile(session);
    uint32 file_id = bak_log_get_id(session, bak->record.data_type, (uint32)point->rst_id, point->asn);
    log_unlock_logfile(session);

    if (file_id != GS_INVALID_ID32) {
        status = build_load_from_online(session, file_id, point, data_size, buf);
        log_unlatch_file(session, file_id);
    } else {
        status = build_load_from_arch(session, point, data_size, buf);
    }

    return status;
}

static status_t bak_build_prepare(knl_session_t *session)
{
    struct st_knl_instance *kernel = session->kernel;
    bak_context_t *ctx = &kernel->backup_ctx;
    bak_t *bak = &ctx->bak;
    bak_remote_t *remote = &bak->remote;
    /*
     * malloc a memory space to store four parts as follows:
     * |--bak->backup_buf--------|--bak->compress_buf------|--remote->send_pack----|--remote->recv_pack----|
     * |--BACKUP_BUFFER_SIZE(bak)--|--BACKUP_BUFFER_SIZE(bak)--|--sizeof(cs_packet_t)--|--sizeof(cs_packet_t)--|
     * so the total space size is BACKUP_BUFFER_SIZE(bak) * 2 and sizeof(cs_packet_t) * 2
     */
    size_t size = BACKUP_BUFFER_SIZE(bak) + BACKUP_BUFFER_SIZE(bak) + sizeof(cs_packet_t) + sizeof(cs_packet_t);
    char *buf = (char *)malloc(size);
    if (buf == NULL) {
        GS_THROW_ERROR(ERR_ALLOC_MEMORY, (uint64)size, "backup");
        return GS_ERROR;
    }

    size_t offset = 0;
    bak->backup_buf = buf + offset;
    offset += BACKUP_BUFFER_SIZE(bak);

    bak_reset_process_ctrl(bak, GS_TRUE);
    bak->compress_buf = buf + offset;
    offset += BACKUP_BUFFER_SIZE(bak);
    /*
     * between the memory addresses of remote->send_pack and bak->backup_buf are bak->backup_buf and bak->compress_buf,
     * so remote->send_pack is offset by 2 * GS_BACKUP_BUFFER_SIZE compared to bak->backup_buf
     */
    remote->send_pack = (cs_packet_t *)(buf + offset);
    offset += sizeof(cs_packet_t);
    /*
     * between the memory addresses of remote->recv_pack and bak->backup_buf are bak->backup_buf,bak->compress_buf and
     * remote->send_pack, so remote->recv_pack is offset by 2 * GS_BACKUP_BUFFER_SIZE and sizeof(cs_packet_t) compared
     * to bak->backup_buf
     */
    remote->recv_pack = (cs_packet_t *)(buf + offset);
    remote->pipe = &remote->send_pipe;
    remote->pipe->link.tcp.sock = CS_INVALID_SOCKET;
    bak->record.start_time = (uint64)cm_now();
    bak->record.attr.backup_type = BACKUP_MODE_FULL;
    bak->is_building = GS_TRUE;

    // if return gs_error, free ctrl_data_buf in bak_build_end
    bak->ctrl_data_buf = (char *)malloc(CTRL_MAX_PAGES(session) * GS_DFLT_CTRL_BLOCK_SIZE);
    if (bak->ctrl_data_buf == NULL) {
        GS_THROW_ERROR(ERR_ALLOC_MEMORY, (uint32)(CTRL_MAX_PAGES(session) * GS_DFLT_CTRL_BLOCK_SIZE), "build");
        return GS_ERROR;
    }

    if (bak->proc_count > BUILD_SINGLE_THREAD) {
        if (cm_aligned_malloc(BACKUP_BUFFER_SIZE(bak), "rst stream buf0", &bak->recv_stream.bufs[0]) != GS_SUCCESS) {
            free(bak->ctrl_data_buf);
            bak->ctrl_data_buf = NULL;
            return GS_ERROR;
        }
        if (cm_aligned_malloc(BACKUP_BUFFER_SIZE(bak), "rst stream buf1", &bak->recv_stream.bufs[1]) != GS_SUCCESS) {
            free(bak->ctrl_data_buf);
            bak->ctrl_data_buf = NULL;
            cm_aligned_free(&bak->recv_stream.bufs[0]);
            return GS_ERROR;
        }
        bak->recv_stream.buf_size = BACKUP_BUFFER_SIZE(bak);
    }

    if (rst_start_write_thread(&ctx->process[BAK_COMMON_PROC]) != GS_SUCCESS) {
        free(bak->ctrl_data_buf);
        bak->ctrl_data_buf = NULL;
        cm_aligned_free(&bak->recv_stream.bufs[0]);
        cm_aligned_free(&bak->recv_stream.bufs[1]);
        return GS_ERROR;
    }

    return GS_SUCCESS;
}

void bak_build_end(knl_session_t *session)
{
    bak_context_t *ctx = &session->kernel->backup_ctx;
    bak_t *bak = &ctx->bak;
    bak_error_t *error_info = &bak->error_info;

    GS_LOG_RUN_INF("[BUILD] remote backup finished, release resources");

    // reset common process finally
    for (int32 i = GS_MAX_BACKUP_PROCESS - 1; i >= 0; i--) {
        bak_reset_process(&ctx->process[i]);
    }

    if (bak->backup_buf != NULL) {
        free(bak->backup_buf);
        bak->backup_buf = NULL;
        bak->depends = NULL;
        bak->compress_buf = NULL;
        bak->ctrl_backup_buf = NULL;
    }

    if (bak->ctrl_data_buf != NULL) {
        free(bak->ctrl_data_buf);
        bak->ctrl_data_buf = NULL;
    }

    /* running on standby, standby decompress file */
    bak_free_compress_context(session, GS_FALSE);

    if (bak->proc_count > BUILD_SINGLE_THREAD) {
        cm_aligned_free(&bak->recv_stream.bufs[0]);
        cm_aligned_free(&bak->recv_stream.bufs[1]);
        bak->recv_stream.buf_size = 0;
    }

    cm_close_device(bak->local.type, &bak->local.handle);
    bak->local.handle = GS_INVALID_HANDLE;

    if (bak->failed && !bak->need_retry) {
        bak_set_fail_error(error_info, "build");
        GS_LOG_RUN_INF("[BUILD] build failed, error code :%d, %s", error_info->err_code, error_info->err_msg);
    }

    build_disconnect(bak);

    bak->failed = GS_FALSE;
    bak->is_building = GS_FALSE;
    bak->record.is_increment = GS_FALSE;
    bak->record.is_repair = GS_FALSE;
    if (rst_delete_track_file(session, bak, GS_TRUE) != GS_SUCCESS) {
        GS_LOG_RUN_WAR("[BUILD] failed to delete bak process file.");
    }

    bak_unset_running(session, ctx);
}

static status_t bak_build_connect(knl_session_t *session)
{
    bak_context_t *ctx = &session->kernel->backup_ctx;
    bak_t *bak = &ctx->bak;
    bak_remote_t *remote = &bak->remote;
    uint16 port;
    char url[GS_HOST_NAME_BUFFER_SIZE + GS_TCP_PORT_MAX_LENGTH + 1] = { 0 };
    char server_host[GS_HOST_NAME_BUFFER_SIZE];
    char bind_host[GS_HOST_NAME_BUFFER_SIZE];
    int32 ret;

    if (lrcv_get_primary_server(session, GS_BACKUP_RETRY_COUNT, server_host, GS_HOST_NAME_BUFFER_SIZE,
        &port) != GS_SUCCESS) {
        return GS_ERROR;
    }

    ret = snprintf_s(url, sizeof(url), sizeof(url) - 1, "%s:%u", server_host, (uint32)port);
    knl_securec_check_ss(ret);
    GS_LOG_RUN_INF("[BUILD] remote backup, standby connect to primary %s", url);

    remote->pipe->connect_timeout = REPL_CONNECT_TIMEOUT;
    remote->pipe->socket_timeout = REPL_SOCKET_TIMEOUT;
    arch_get_bind_host(session, server_host, bind_host, GS_HOST_NAME_BUFFER_SIZE);
    if (cs_connect(url, remote->pipe, bind_host, NULL, NULL) != GS_SUCCESS) {
        GS_LOG_RUN_ERR("[BACKUP] failed to connect %s", url);
        return GS_ERROR;
    }

    if (knl_login(session, remote->pipe, REP_LOGIN_BACKUP, (const char *)bind_host, NULL) != GS_SUCCESS) {
        GS_LOG_RUN_ERR("[BACKUP] login failed");
        return GS_ERROR;
    }

    return GS_SUCCESS;
}

static status_t bak_build_start(knl_session_t *session)
{
    bak_context_t *ctx = &session->kernel->backup_ctx;
    bak_t *bak = &ctx->bak;

    if (arch_init(session) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (bak_alloc_compress_context(session, GS_FALSE) != GS_SUCCESS) {
        bak->failed = GS_TRUE;
        return GS_ERROR;
    }

    if (bak_build_prepare(session) != GS_SUCCESS) {
        bak->failed = GS_TRUE;
        return GS_ERROR;
    }

    for (uint32 i = 0; i < BUILD_MAX_RETRY; i++) {
        if (bak_build_connect(session) == GS_SUCCESS) {
            return GS_SUCCESS;
        }
    }

    bak->need_retry = GS_TRUE;
    return GS_ERROR;
}

static bak_file_type_t rst_calc_file_type(uint32 type)
{
    bak_file_type_t file_type;

    switch (type) {
        case BAK_MSG_TYPE_CTRL:
            file_type = BACKUP_CTRL_FILE;
            break;
        case BAK_MSG_TYPE_DATA:
            file_type = BACKUP_DATA_FILE;
            break;
        case BAK_MSG_TYPE_ARCH:
            file_type = BACKUP_ARCH_FILE;
            break;
        case BAK_MSG_TYPE_LOG:
            file_type = BACKUP_LOG_FILE;
            break;
        default:
            file_type = BACKUP_HEAD_FILE;
            break;
    }

    return file_type;
}

static inline bool32 rst_restart_from_file_end(bak_t *bak, bak_start_msg_t *start_msg)
{
    return (BAK_IS_FULL_BUILDING(bak) && bak->need_check && bak->file_count != start_msg->curr_file_index);
}

static status_t rst_read_start_msg(knl_session_t *session, bak_start_msg_t *start_msg)
{
    bak_t *bak = &session->kernel->backup_ctx.bak;
    bool32 read_end = GS_FALSE;

    if (rst_agent_read_head(bak, BAK_PKG_FILE_START, &bak->remote.remain_data_size, &read_end) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (bak_agent_recv(bak, (char *)start_msg, sizeof(bak_start_msg_t)) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (bak_agent_send_pkg(bak, BAK_PKG_ACK) != GS_SUCCESS) {
        return GS_ERROR;
    }
    GS_LOG_RUN_INF("[RESTORE]: type:%u, file_id:%u, curr_file_index:%u", start_msg->type, start_msg->file_id,
        start_msg->curr_file_index);
    if (start_msg->type != BAK_MSG_TYPE_HEAD) {
        if (rst_restart_from_file_end(bak, start_msg)) {
            GS_LOG_RUN_INF("[BUILD] break-point build restart from end of file [%u/%u]", bak->file_count,
                start_msg->curr_file_index);
            if (bak->file_count != start_msg->curr_file_index - 1) {
                GS_LOG_RUN_INF("[BUILD] break-point build restart from unexpected position");
                return GS_ERROR;
            }
            bak->file_count++;
        }
        bak->need_check = GS_FALSE;
        bak->files[bak->file_count].type = rst_calc_file_type(start_msg->type);
        bak->files[bak->file_count].id = start_msg->file_id;
        bak->curr_file_index = bak->file_count;
        bak->file_count++;
    }

    return GS_SUCCESS;
}

static bak_stage_t rst_calc_stage(uint32 type)
{
    bak_stage_t stage;

    switch (type) {
        case BAK_MSG_TYPE_CTRL:
            stage = BACKUP_CTRL_STAGE;
            break;
        case BAK_MSG_TYPE_DATA:
            stage = BACKUP_DATA_STAGE;
            break;
        case BAK_MSG_TYPE_HEAD:
            stage = BACKUP_HEAD_STAGE;
            break;
        default:
            stage = BACKUP_LOG_STAGE;
            break;
    }

    return stage;
}

static status_t rst_build_receive_files(knl_session_t *session)
{
    bak_t *bak = &session->kernel->backup_ctx.bak;
    bak_start_msg_t start_msg;
    bak_stage_t stage = BACKUP_CTRL_STAGE;

    bak->head_is_built = GS_FALSE;
    while (!bak->failed && stage != BACKUP_HEAD_STAGE) {
        if (bak_check_session_status(session) != GS_SUCCESS) {
            return GS_ERROR;
        }

        if (rst_read_start_msg(session, &start_msg) != GS_SUCCESS) {
            return GS_ERROR;
        }

        stage = rst_calc_stage(start_msg.type);
        if (stage == BACKUP_HEAD_STAGE) {
            bak->remote.remain_data_size = 0;
            bak->progress.stage = BACKUP_READ_FINISHED;
            break;
        }

        if (bak->progress.stage != stage) {
            bak->progress.stage = stage;
        }

        bak_file_type_t type = rst_calc_file_type(start_msg.type);
        if (type >= BACKUP_DATA_FILE && type <= BACKUP_ARCH_FILE && bak_paral_task_enable(session)) {
            if (rst_stream_read_file(session) != GS_SUCCESS) {
                return GS_ERROR;
            }
        } else {
            if (rst_read_file(session, bak->curr_file_index) != GS_SUCCESS) {
                return GS_ERROR;
            }
        }

        if (bak_agent_send_pkg(bak, BAK_PKG_ACK) != GS_SUCCESS) {
            return GS_ERROR;
        }
    }

    return bak->failed ? GS_ERROR : GS_SUCCESS;
}

static status_t rst_restore_keyfile(knl_session_t *session, const char *buf, uint32 buf_size, char *file_name)
{
    int32 handle = INVALID_FILE_HANDLE;

    if (cm_file_exist(file_name)) {
        if (cm_remove_file(file_name) != GS_SUCCESS) {
            GS_LOG_RUN_ERR("[RST] failed to remove key file %s", file_name);
            return GS_ERROR;
        }
    }

    if (cm_open_file(file_name, O_RDWR | O_CREAT | O_SYNC, &handle) != GS_SUCCESS) {
        return GS_ERROR;
    }
    if (cm_chmod_file(S_IRUSR | S_IWUSR, handle) != GS_SUCCESS) {
        cm_close_file(handle);
        GS_LOG_RUN_ERR("failed to modify key file %s permissions", file_name);
        return GS_ERROR;
    }

    int64 file_size = cm_file_size(handle);
    if (file_size < 0) {
        cm_close_file(handle);
        GS_LOG_RUN_ERR("[RST] file seek fail %s", file_name);
        return GS_ERROR;
    }

    file_size = ((keyfile_ctrl_t *)buf)->size;
    knl_panic(sizeof(keyfile_ctrl_t) + file_size <= GS_KMC_MAX_KEY_SIZE);

    if (cm_write_file(handle, buf + sizeof(keyfile_ctrl_t), (int32)file_size) != GS_SUCCESS) {
        cm_close_file(handle);
        GS_LOG_RUN_ERR("[RST] fail to write file %s", file_name);
        return GS_ERROR;
    }
    cm_close_file(handle);

    return GS_SUCCESS;
}

static status_t rst_build_keyfile(knl_session_t *session, bak_head_t *head, int32 read_size)
{
    int64 file_size = ((keyfile_ctrl_t *)head)->size;
    if (file_size <= 0) {
        GS_LOG_RUN_INF("[BUILD] build without ssl, won't reset keyfile");
        return GS_SUCCESS;
    }

    for (uint32 i = 0; i < GS_KMC_MAX_KEYFILE_NUM; i++) {
        if (rst_restore_keyfile(session, (char *)head, (uint32)read_size,
            session->kernel->attr.kmc_key_files[i].name) != GS_SUCCESS) {
            return GS_ERROR;
        }
    }

    if (cm_kmc_reset() != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (g_knl_callback.sysdba_privilege() != GS_SUCCESS) {
        return GS_ERROR;
    }
    return GS_SUCCESS;
}

static status_t rst_build_head(knl_session_t *session)
{
    bak_context_t *ctx = &session->kernel->backup_ctx;
    bak_t *bak = &ctx->bak;
    int32 read_size;
    bool32 read_end = GS_FALSE;
    int32 buf_size = sizeof(bak_head_t);
    bak_head_t *head = (bak_head_t *)bak->backup_buf;
    int32 expect_size = (int32)sizeof(bak_head_t);

    if (bak->is_building) {
        buf_size += GS_KMC_MAX_KEY_SIZE;
        expect_size += GS_KMC_MAX_KEY_SIZE;
    }

    if (rst_read_data(ctx, (void *)head, buf_size, &read_size, &read_end, 0) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (rst_read_check_size(read_size, expect_size, "backupset head") != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (bak->is_building) {
        if (rst_build_keyfile(session, head, read_size) != GS_SUCCESS) {
            GS_LOG_RUN_ERR("[BUILD] rst build keyfile failed.");
            return GS_ERROR;
        }
        head = (bak_head_t *)((char *)head + GS_KMC_MAX_KEY_SIZE);
    }

    if (rst_set_head(session, head, GS_FALSE) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (rst_wait_agent_process(bak) != GS_SUCCESS) {
        return GS_ERROR;
    }
    bak->head_is_built = GS_TRUE;

    return GS_SUCCESS;
}

static status_t build_analyze_entry(knl_session_t *session, log_entry_t *log)
{
    if (RD_TYPE_IS_ENTER_PAGE(log->type)) {
        rd_enter_page_t *redo = (rd_enter_page_t *)log->data;
        page_id_t page_id = MAKE_PAGID(redo->file, redo->page);
        if (build_aly_add_page(session, page_id) != GS_SUCCESS) {
            return GS_ERROR;
        }
    }
    return GS_SUCCESS;
}

static status_t build_analyze_group(knl_session_t *session, log_group_t *group)
{
    uint32 offset = sizeof(log_group_t);
    log_entry_t *log = NULL;

    while (offset < group->size) {
        log = (log_entry_t *)((char *)group + offset);
        if (build_analyze_entry(session, log) != GS_SUCCESS) {
            return GS_ERROR;
        }
        offset += log->size;
    }
    return GS_SUCCESS;
}

static status_t build_analyze_batch(knl_session_t *session, log_batch_t *batch)
{
    knl_session_t *se = session->kernel->sessions[SESSION_ID_KERNEL];
    log_context_t *ctx = &session->kernel->redo_ctx;
    log_cursor_t cursor;

    rcy_init_log_cursor(&cursor, batch);

    log_group_t *group = log_fetch_group(ctx, &cursor);
    while (group != NULL) {
        if (build_analyze_group(se, group) != GS_SUCCESS) {
            return GS_ERROR;
        }

        group = log_fetch_group(ctx, &cursor);
    }

    return GS_SUCCESS;
}

static inline void build_next_file(log_point_t *point, bool32 *need_more_log)
{
    point->asn++;
    point->block_id = 0;
    *need_more_log = GS_TRUE;
}

static bool32 build_check_batch(log_point_t *point, log_batch_t *batch, bool32 *need_more_log)
{
    if (batch->head.magic_num != LOG_MAGIC_NUMBER || !LFN_IS_CONTINUOUS(batch->head.point.lfn, point->lfn)) {
        build_next_file(point, need_more_log);
        return GS_FALSE;
    }

    return GS_TRUE;
}

static status_t build_analyze(knl_session_t *session, log_point_t *point, log_batch_t *batch, uint32 block_size,
    uint32 data_size_input, bool32 *need_more_log)
{
    uint32 data_size = data_size_input;
    log_context_t *ctx = &session->kernel->redo_ctx;
    log_batch_tail_t *tail = (log_batch_tail_t *)((char *)batch + batch->size - sizeof(log_batch_tail_t));
    bool32 first_batch = GS_TRUE;
    if (data_size == 0) {
        build_next_file(point, need_more_log);
        return GS_SUCCESS;
    }

    while (data_size >= sizeof(log_batch_id_t)) {
        if (!build_check_batch(point, batch, need_more_log)) {
            return GS_SUCCESS;
        }
        if (data_size < batch->space_size) {
            if (first_batch) {
                *need_more_log = GS_FALSE;
                GS_LOG_RUN_ERR("[BUILD] analyze failed, invalid batch(%u,%u).", point->asn, point->block_id);
                return GS_ERROR;
            }
            *need_more_log = GS_TRUE;
            return GS_SUCCESS;
        }

        if (!rcy_validate_batch(batch, tail)) {
            *need_more_log = GS_FALSE;
            GS_LOG_RUN_ERR("[BUILD] analyze failed, invalid batch(%u,%u).", point->asn, point->block_id);
            return GS_ERROR;
        }
        first_batch = GS_FALSE;
        if (rcy_verify_checksum(session, batch) != GS_SUCCESS) {
            *need_more_log = GS_FALSE;
            GS_LOG_RUN_ERR("[BUILD] analyze failed");
            return GS_ERROR;
        }

        if (batch->encrypted) {
            if (log_decrypt(session, batch, ctx->logwr_cipher_buf, ctx->logwr_cipher_buf_size) != GS_SUCCESS) {
                *need_more_log = GS_FALSE;
                GS_LOG_RUN_ERR("[BUILD] analyze failed");
                return GS_ERROR;
            }
        }

        if (build_analyze_batch(session, batch) != GS_SUCCESS) {
            *need_more_log = GS_FALSE;
            GS_LOG_RUN_ERR("[BUILD] analyze failed");
            return GS_ERROR;
        }

        data_size -= batch->space_size;
        point->lfn = batch->head.point.lfn;

        point->block_id += batch->space_size / block_size;
        batch = (log_batch_t *)((char *)batch + batch->space_size);
        tail = (log_batch_tail_t *)((char *)batch + batch->size - sizeof(log_batch_tail_t));
    }

    *need_more_log = GS_TRUE;
    return GS_SUCCESS;
}

static status_t build_analyze_redo(knl_session_t *session, log_point_t *curr_point)
{
    aligned_buf_t log_buf  = session->kernel->backup_ctx.bak.log_buf;
    log_batch_t *batch = NULL;
    bool32 need_more_log = GS_FALSE;
    uint32 data_size = 0;
    uint32 block_size;

    while (bak_load_log_batch(session, curr_point, &data_size, &log_buf, &block_size) == GS_SUCCESS) {
        batch = (log_batch_t *)log_buf.aligned_buf;

        if (build_analyze(session, curr_point, batch, block_size, data_size, &need_more_log) != GS_SUCCESS) {
            cm_aligned_free(&log_buf);
            GS_THROW_ERROR(ERR_INVALID_OPERATION, ", analyze log failed");
            return GS_ERROR;
        }

        if (!need_more_log) {
            break;
        }
    }

    return GS_SUCCESS;
}

static status_t build_repair_prepare(knl_session_t *session, uint64 *base_lsn)
{
    reset_log_t resetlog = session->kernel->lrcv_ctx.primary_resetlog;
    aligned_buf_t log_buf = session->kernel->backup_ctx.bak.log_buf;
    log_point_t target_point;
    uint32 data_size = 0;
    log_point_t *lrp = &dtc_my_ctrl(session)->lrp_point;

    target_point.rst_id = resetlog.rst_id - 1;
    target_point.asn = resetlog.last_asn;
    target_point.lfn = resetlog.last_lfn;
    target_point.block_id = 0;

    GS_LOG_RUN_INF("[REPAIR] analysis primary resetlog rstid-asn-lfn: %u-%u-%llu", resetlog.rst_id, resetlog.last_asn,
        resetlog.last_lfn);
    if (log_load(session) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (build_load_batch(session, &target_point, &data_size, &log_buf) != GS_SUCCESS) {
        GS_THROW_ERROR(ERR_INVALID_OPERATION, ", can not find resetlog point in standby.");
        return GS_ERROR;
    }

    if (data_size != 0) {
        log_batch_t *batch = (log_batch_t *)log_buf.aligned_buf;
        *base_lsn = rcy_fetch_batch_lsn(session, batch);
        target_point.block_id = batch->head.point.block_id;
    } else {
        GS_THROW_ERROR(ERR_INVALID_OPERATION, ", can not find resetlog point in standby.");
        return GS_ERROR;
    }

    target_point.lfn -= 1;

    GS_LOG_RUN_INF("[REPAIR] analysis start point rstid-asn-lfn: %u-%u-%llu", target_point.rst_id, target_point.asn,
        (uint64)target_point.lfn);
    if (build_analyze_redo(session, &target_point) != GS_SUCCESS) {
        return GS_ERROR;
    }
    GS_LOG_RUN_INF("[REPAIR] analysis end point rstid-asn-lfn: %u-%u-%llu", target_point.rst_id, target_point.asn,
        (uint64)target_point.lfn);

    if (target_point.lfn < lrp->lfn) {
        GS_THROW_ERROR(ERR_INVALID_OPERATION, ", analysis end lfn is smaller than lrp lfn.");
        return GS_ERROR;
    }
    return GS_SUCCESS;
}

static status_t rst_build_get_base_lsn(knl_session_t *session, uint64 *base_lsn)
{
    log_point_t start_point = dtc_my_ctrl(session)->rcy_point;
    log_batch_t *batch = NULL;
    log_batch_tail_t *tail = NULL;
    uint32 data_size = 0;
    bak_context_t *ctx = &session->kernel->backup_ctx;
    bak_t *bak = &ctx->bak;
    database_t *db = &session->kernel->db;
    reset_log_t rst_log = db->ctrl.core.resetlogs;
    aligned_buf_t log_buf;
    uint32 block_size;

    if (log_load(session) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (dtc_my_ctrl(session)->rcy_point.lfn == dtc_my_ctrl(session)->lrp_point.lfn) {
        *base_lsn = DB_CURR_LSN(session);
        return GS_SUCCESS;
    }

    if (cm_aligned_malloc(GS_MAX_BATCH_SIZE, "backup log buffer", &log_buf) != GS_SUCCESS) {
        return GS_ERROR;
    }

    for (;;) {
        if (bak_load_log_batch(session, &start_point, &data_size, &log_buf, &block_size) != GS_SUCCESS) {
            cm_aligned_free(&log_buf);
            return GS_ERROR;
        }

        batch = (log_batch_t *)log_buf.aligned_buf;
        if (data_size >= sizeof(log_batch_t) && data_size >= batch->size) {
            tail = (log_batch_tail_t *)((char *)batch + batch->size - sizeof(log_batch_tail_t));
            if (rcy_validate_batch(batch, tail)) {
                break;
            }
        }

        start_point.asn++;
        start_point.rst_id = bak_get_rst_id(bak, start_point.asn, &(rst_log));
        start_point.block_id = 0;
    }

    *base_lsn = rcy_fetch_batch_lsn(session, batch);
    cm_aligned_free(&log_buf);
    return GS_SUCCESS;
}

bool32 rst_logfile_not_changed(knl_instance_t *kernel, log_file_ctrl_t *log_old, log_file_ctrl_t *log_new)
{
    char name[GS_FILE_NAME_BUFFER_SIZE];
    errno_t ret;

    ret = strcpy_sp(name, GS_FILE_NAME_BUFFER_SIZE, log_new->name);
    knl_securec_check(ret);

    if (LOG_IS_DROPPED(log_old->flg) != LOG_IS_DROPPED(log_new->flg)) {
        GS_LOG_RUN_ERR("[BUILD] primary's log flag has changed old: %u, new: %u", log_old->flg, log_new->flg);
        return GS_FALSE;
    }
    if (db_change_storage_path(&kernel->attr.log_file_convert, name, GS_FILE_NAME_BUFFER_SIZE) != GS_SUCCESS) {
        return GS_FALSE;
    }
    if (!cm_str_equal(log_old->name, name)) {
        GS_LOG_RUN_ERR("[BUILD] primary's log amount/name has changed old: %s, new: %s", log_old->name, name);
        return GS_FALSE;
    }

    return GS_TRUE;
}

bool32 rst_datafile_not_changed(knl_instance_t *kernel, datafile_ctrl_t *df_old, datafile_ctrl_t *df_new)
{
    char name[GS_FILE_NAME_BUFFER_SIZE];
    errno_t ret;

    ret = strcpy_sp(name, GS_FILE_NAME_BUFFER_SIZE, df_new->name);
    knl_securec_check(ret);

    if (df_old->used != df_new->used || (df_old->flag & DATAFILE_FLAG_ONLINE) != (df_new->flag & DATAFILE_FLAG_ONLINE)) {
        GS_LOG_RUN_ERR("[BUILD] primary's data flag has changed old_used: %u, new_used: %u, old_flag: %u,"
            "new_flag: %u", df_old->used, df_new->used, df_old->flag, df_new->flag);
        return GS_FALSE;
    }

    if (df_old->create_version != df_new->create_version) {
        GS_LOG_RUN_ERR("[BUILD] primary's data version has changed old: %u, new: %u", df_old->create_version,
            df_new->create_version);
        return GS_FALSE;
    }
    if (db_change_storage_path(&kernel->attr.data_file_convert, name, GS_FILE_NAME_BUFFER_SIZE) != GS_SUCCESS) {
        return GS_FALSE;
    }
    if (!cm_str_equal(df_old->name, name)) {
        GS_LOG_RUN_ERR("[BUILD] primary's data amount/name has changed old: %s, new: %s", df_old->name, name);
        return GS_FALSE;
    }

    return GS_TRUE;
}


bool32 rst_db_files_not_changed(knl_session_t *session, ctrl_page_t *new_ctrl)
{
    knl_instance_t *kernel = session->kernel;
    datafile_ctrl_t *df_old = NULL;
    datafile_ctrl_t *df_new = NULL;
    log_file_ctrl_t *log_old = NULL;
    log_file_ctrl_t *log_new = NULL;
    uint32 offset = CTRL_LOG_SEGMENT;
    uint32 count;
    uint32 i;

    for (i = 0; i < GS_MAX_LOG_FILES; i++) {
        log_new = (log_file_ctrl_t *)db_get_ctrl_item(new_ctrl, i, sizeof(log_file_ctrl_t), offset);
        log_old = MY_LOGFILE_SET(session)->items[i].ctrl;
        if (rst_logfile_not_changed(kernel, log_old, log_new) != GS_TRUE) {
            return GS_FALSE;
        }
    }

    count = CTRL_MAX_BUF_SIZE / sizeof(log_file_ctrl_t);
    offset = offset + (GS_MAX_LOG_FILES - 1) / count + 1;

    count = CTRL_MAX_BUF_SIZE / sizeof(space_ctrl_t);
    offset = offset + (GS_MAX_SPACES - 1) / count + 1;

    for (i = 0; i < GS_MAX_DATA_FILES; i++) {
        df_new = (datafile_ctrl_t *)db_get_ctrl_item(new_ctrl, i, sizeof(datafile_ctrl_t), offset);
        df_old = kernel->db.datafiles[i].ctrl;
        if (rst_datafile_not_changed(kernel, df_old, df_new) != GS_TRUE) {
            return GS_FALSE;
        }
    }

    return GS_TRUE;
}

status_t rst_build_update_logfiles(knl_session_t *session, bak_process_t *ctx)
{
    log_file_ctrl_t *logfile = NULL;
    int32 *handle = NULL;
    bool32 is_dbstor = knl_dbs_is_enable_dbs();

    for (uint32 i = 0; i < dtc_my_ctrl(session)->log_hwm; i++) {
        logfile = MY_LOGFILE_SET(session)->items[i].ctrl;
        handle = &MY_LOGFILE_SET(session)->items[i].handle;
        if (LOG_IS_DROPPED(logfile->flg)) {
            if (strlen(ctx->logfile_name[i]) == 0) {
                continue;
            }

            cm_close_device(ctx->log_type[i], handle);
            if (cm_remove_device(ctx->log_type[i], ctx->logfile_name[i]) != GS_SUCCESS) {
                return GS_ERROR;
            }
            continue;
        }

        if (cm_exist_device(logfile->type, logfile->name) && *handle != GS_INVALID_HANDLE) {
            cm_close_device(ctx->log_type[i], handle);
            if (cm_open_device(logfile->name, logfile->type, knl_redo_io_flag(session), handle) != GS_SUCCESS) {
                GS_LOG_RUN_ERR("[RESTORE] failed to open %s ", logfile->name);
                return GS_ERROR;
            }
            continue;
        }

        if (strlen(ctx->logfile_name[i]) > 0) {
            cm_close_device(ctx->log_type[i], handle);
            if (cm_remove_device(ctx->log_type[i], ctx->logfile_name[i]) != GS_SUCCESS) {
                return GS_ERROR;
            }
        }

        if (cm_create_device(logfile->name, logfile->type, knl_redo_io_flag(session), handle) != GS_SUCCESS) {
            GS_LOG_RUN_ERR("[RESTORE] failed to create %s ", logfile->name);
            return GS_ERROR;
        }
        if (is_dbstor) {
            break;
        }
    }

    return GS_SUCCESS;
}

static status_t rst_load_ctrlfile(knl_session_t *session)
{
    bak_context_t *ctx = &session->kernel->backup_ctx;
    bak_t *bak = &ctx->bak;
    text_t ctrlfiles;
    database_t *db = &session->kernel->db;
    dtc_node_ctrl_t *node = dtc_my_ctrl(session);

    char *param = cm_get_config_value(session->kernel->attr.config, "CONTROL_FILES");
    cm_str2text(param, &ctrlfiles);
    if (ctrlfiles.len == 0) {
        GS_LOG_RUN_INF("[BUILD] ctrl file does not exist");
        GS_THROW_ERROR(ERR_LOAD_CONTROL_FILE, "ctrl file does not exist");
        return GS_ERROR;
    }

    if (db_load_ctrlspace(session, &ctrlfiles) != GS_SUCCESS) {
        return GS_ERROR;
    }

    node->archived_start = 0;
    node->archived_end = 0;
    db->ctrl.core.build_completed = GS_FALSE;
    db->ctrl.core.db_role = bak->is_building ? REPL_ROLE_PHYSICAL_STANDBY : REPL_ROLE_PRIMARY;
    bak->ctrlfile_completed = GS_TRUE;

    return GS_SUCCESS;
}

static status_t rst_check_ctrl_file(knl_session_t *session)
{
    bak_context_t *ctx = &session->kernel->backup_ctx;
    bak_t *bak = &ctx->bak;
    bool32 read_end = GS_FALSE;
    int32 read_size;

    if (!BAK_IS_FULL_BUILDING(bak) || bak->is_first_link) {
        return GS_SUCCESS;
    }

    GS_LOG_RUN_INF("[BUILD] start read ctrl file");
    if (rst_read_data(ctx, bak->ctrl_data_buf, CTRL_MAX_PAGES(session) * GS_DFLT_CTRL_BLOCK_SIZE,
                      &read_size, &read_end, 0) != GS_SUCCESS) {
        return GS_ERROR;
    }
    GS_LOG_RUN_INF("[BUILD] ctrl file read_size : %u", read_size);

    if (bak_agent_send_pkg(bak, BAK_PKG_ACK) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (rst_load_ctrlfile(session) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (!db_sysdata_version_is_equal(session, GS_FALSE)) {
        return GS_ERROR;
    }

    GS_LOG_RUN_INF("[BUILD] start check file amount/name");
    if (!rst_db_files_not_changed(session, (ctrl_page_t *)bak->ctrl_data_buf)) {
        GS_LOG_RUN_ERR("[BUILD] primary's log/data amount/name has changed");
        if (rst_delete_track_file(session, bak, GS_FALSE) != GS_SUCCESS) {
            return GS_ERROR;
        }

        GS_THROW_ERROR(ERR_INVALID_OPERATION, ": primary's log/data amount/name has changed");
        return GS_ERROR;
    }
    GS_LOG_RUN_INF("[BUILD] check file amount/name successfully");

    return GS_SUCCESS;
}

static status_t rst_build_restore(knl_session_t *session)
{
    if (rst_restore_config_param(session) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (rst_check_ctrl_file(session) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (rst_build_receive_files(session) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (rst_build_head(session) != GS_SUCCESS) {
        return GS_ERROR;
    }

    return GS_SUCCESS;
}

static char *bak_get_compress_name(compress_algo_e compress)
{
    switch (compress) {
        case COMPRESS_NONE:
            return "NONE";
        case COMPRESS_ZLIB:
            return "ZLIB";
        case COMPRESS_ZSTD:
            return "ZSTD";
        case COMPRESS_LZ4:
            return "LZ4";
        default:
            cm_assert(GS_FALSE);
            return "NULL";
    }
}

bool32 bak_parameter_is_valid(build_progress_t *build_progress)
{
    uint32 stage = bak_get_build_stage(&build_progress->stage);
    if (stage != BUILD_LOG_STAGE && stage != BUILD_DATA_STAGE) {
        return GS_FALSE;
    }

    return GS_TRUE;
}

static status_t bak_obtain_build_progress(bak_t *bak, bak_process_t *proc, const char *file_name)
{
    build_progress_t *build_progress = &bak->progress.build_progress;
    int32 handle = GS_INVALID_HANDLE;

    if (cm_open_file(file_name, O_BINARY | O_RDWR, &handle) != GS_SUCCESS) {
        GS_LOG_RUN_ERR("[BAK] failed to open %s ", file_name);
        return GS_ERROR;
    }

    if (cm_read_file(handle, build_progress, sizeof(build_progress_t), NULL) != GS_SUCCESS) {
        cm_close_file(handle);
        GS_LOG_RUN_ERR("[BAK] failed to read %s ", file_name);
        return GS_ERROR;
    }
    cm_close_file(handle);
    GS_LOG_RUN_INF("[BUILD] write build progress [%u/%u/%llu/%u/%u/%u/%d]", build_progress->stage,
                   build_progress->file_id, build_progress->data_offset, build_progress->asn,
                   build_progress->curr_file_index, build_progress->last_file_index, build_progress->start_time);

    if (!bak_parameter_is_valid(build_progress)) {
        GS_LOG_RUN_ERR("[BAK] parameters are not effective in %s ", file_name);
        return GS_ERROR;
    }

    bak->file_count = build_progress->curr_file_index;

    return GS_SUCCESS;
}

static status_t bak_agent_send_build_ctrl(knl_session_t *session, bak_t *bak, bak_process_t *proc)
{
    bak_agent_head_t head;
    bak_attr_t *attr = &bak->record.attr;
    build_param_ctrl_t ctrl;
    build_progress_t *build_progress = &bak->progress.build_progress;
    char bak_process_file[GS_FILE_NAME_BUFFER_SIZE];
    errno_t ret;
    int32 primary_start_time;
    bool32 read_end = GS_FALSE;
    int32 read_size;

    head.ver = BAK_AGENT_PROTOCOL;
    head.serial_number = bak->remote.serial_number++;
    head.cmd = BAK_PKG_START;
    head.len = sizeof(bak_agent_head_t) + sizeof(build_param_ctrl_t) + sizeof(build_progress_t);
    head.flags = 0;
    head.reserved = 0;
    ctrl.compress = attr->compress;
    ctrl.compress_level = bak->compress_ctx.compress_level;
    ctrl.parallelism = bak->proc_count;
    ctrl.is_increment = bak->record.is_increment;
    ctrl.base_lsn = attr->base_lsn;
    ctrl.is_repair = bak->record.is_repair;
    ctrl.buffer_size = bak->backup_buf_size;

    GS_LOG_DEBUG_INF("[BACKUP] send type %d", head.cmd);
    // send head
    if (bak_agent_send(bak, (char *)&head, sizeof(bak_agent_head_t)) != GS_SUCCESS) {
        return GS_ERROR;
    }
    // send compress param
    if (bak_agent_send(bak, (char *)&ctrl, sizeof(build_param_ctrl_t)) != GS_SUCCESS) {
        return GS_ERROR;
    }
    // send break-point info if necessary
    ret = snprintf_s(bak_process_file, GS_FILE_NAME_BUFFER_SIZE, GS_FILE_NAME_BUFFER_SIZE - 1, "%s/bak_process_file",
        session->kernel->home);
    knl_securec_check_ss(ret);
    if (cm_file_exist((const char *)bak_process_file)) {
        if (bak_obtain_build_progress(bak, proc, bak_process_file) != GS_SUCCESS) {
            GS_THROW_ERROR(ERR_INVALID_OPERATION, ", building process file is not effective.");
            return GS_ERROR;
        }
        bak->need_check = GS_TRUE;
        bak->is_first_link = GS_FALSE;
    } else {
        ret = memset_sp(build_progress, sizeof(build_progress_t), 0, sizeof(build_progress_t));
        knl_securec_check(ret);
        bak->is_first_link = GS_TRUE;
    }

    if (bak_agent_send(bak, (char *)build_progress, sizeof(build_progress_t)) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (rst_agent_read(bak, (char *)&primary_start_time, sizeof(int32), &read_size, &read_end) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (bak->is_first_link) {
        build_progress->start_time = primary_start_time;
        GS_LOG_RUN_INF("[BUILD] first link [primary/standby] start time : [%d/%d]", primary_start_time,
            build_progress->start_time);
    } else {
        if (build_progress->start_time != primary_start_time) {
            GS_LOG_RUN_INF("[BUILD] [primary/standby] start time : [%d/%d]", primary_start_time,
                build_progress->start_time);
            GS_LOG_RUN_INF("[BUILD] primary has restart or timeout, only non-breakpoint building is allowed");
            if (rst_delete_track_file(session, bak, GS_FALSE) != GS_SUCCESS) {
                return GS_ERROR;
            }

            GS_THROW_ERROR(ERR_INVALID_OPERATION,
                "primary has restart or timeout, only non-breakpoint building is allowed");
            return GS_ERROR;
        }
        GS_LOG_RUN_INF("[BUILD] [primary/standby] start time : [%d/%d]", primary_start_time,
            build_progress->start_time);
    }

    return GS_SUCCESS;
}

static status_t rst_build_prepare_basefiles(knl_session_t *session, bak_process_t *ctx)
{
    knl_instance_t *kernel = session->kernel;
    datafile_t *df = NULL;
    log_file_ctrl_t *log = NULL;
    uint32 i;
    errno_t ret;

    for (i = 0; i < GS_MAX_DATA_FILES; i++) {
        ctx->datafile_size[i] = GS_INVALID_INT64;
        df = &kernel->db.datafiles[i];
        if (!df->ctrl->used || !DATAFILE_IS_ONLINE(df)) {
            ctx->datafile_name[i][0] = '\0';
            continue;
        }

        if (cm_open_device(df->ctrl->name, df->ctrl->type, knl_io_flag(session), &ctx->datafiles[i]) != GS_SUCCESS) {
            GS_LOG_RUN_ERR("[BUILD] failed to open %s ", df->ctrl->name);
            return GS_ERROR;
        }

        ret = strcpy_sp(ctx->datafile_name[i], GS_FILE_NAME_BUFFER_SIZE, df->ctrl->name);
        knl_securec_check(ret);
        ctx->file_type[i] = df->ctrl->type;
        ctx->datafile_version[i] = df->ctrl->create_version;

        ctx->datafile_size[i] = cm_device_size(cm_device_type(ctx->datafile_name[i]), ctx->datafiles[i]);
        if (ctx->datafile_size[i] == -1) {
            GS_THROW_ERROR(ERR_SEEK_FILE, 0, SEEK_END, errno);
            return GS_ERROR;
        }
    }

    for (i = 0; i < GS_MAX_LOG_FILES; i++) {
        ctx->logfile_name[i][0] = '\0';
        ctx->log_type[i] = 0;
    }

    for (i = 0; i < MY_LOGFILE_SET(session)->logfile_hwm; i++) {
        log = MY_LOGFILE_SET(session)->items[i].ctrl;
        if (LOG_IS_DROPPED(log->flg)) {
            continue;
        }

        ret = strcpy_sp(ctx->logfile_name[i], GS_FILE_NAME_BUFFER_SIZE, log->name);
        knl_securec_check(ret);
        ctx->log_type[i] = log->type;
    }

    return GS_SUCCESS;
}

static status_t bak_check_keep_alive(knl_session_t *session)
{
    bak_context_t *ctx = &session->kernel->backup_ctx;
    uint32 build_keep_alive_timeout = session->kernel->attr.build_keep_alive_timeout;
    bak_t *bak = &ctx->bak;

    if (bak->record.attr.level != 0) {
        return GS_SUCCESS;
    }

    // will not retry while timeout
    bak->build_retry_time++;
    if (bak->build_retry_time == build_keep_alive_timeout) {
        bak->need_retry = GS_FALSE;
        GS_THROW_ERROR(ERR_INVALID_OPERATION, ", please check run log for information of building timeout.");
        return GS_ERROR;
    }

    return GS_SUCCESS;
}

static status_t bak_agent_send_diff_pages(knl_session_t *session)
{
    bak_t *bak = &session->kernel->backup_ctx.bak;
    bak_agent_head_t head;

    uint32 page_count = BUILD_ALY_MAX_ITEM - bak->build_aly_free_list.count;
    knl_panic(page_count == bak->page_count);

    GS_LOG_RUN_INF("[BUILD] start to send page list to primary, page_count: %u", page_count);
    head.ver = BAK_AGENT_PROTOCOL;
    head.cmd = BAK_PKG_DATA;
    head.len = sizeof(bak_agent_head_t) + page_count * sizeof(page_id_t);
    head.serial_number = 0;
    head.flags = 0;
    head.reserved = 0;

    if (bak_agent_send(bak, (char *)&head, sizeof(bak_agent_head_t)) != GS_SUCCESS) {
        return GS_ERROR;
    }
    if (bak_agent_send(bak, (char *)bak->build_aly_pages, page_count * sizeof(page_id_t)) != GS_SUCCESS) {
        return GS_ERROR;
    }
    if (bak_agent_wait_pkg(bak, BAK_PKG_ACK) != GS_SUCCESS) {
        return GS_ERROR;
    }
    GS_LOG_RUN_INF("[BUILD] send page list to primary successfully");
    return GS_SUCCESS;
}

inline static void bak_brain_close_file(int32 *handle, uint32 count)
{
    for (uint32 i = 0; i < count; i++) {
        cm_close_file(handle[i]);
    }
}

static status_t bak_brain_reset_page(knl_session_t *session)
{
    int32 handles[GS_MAX_DATA_FILES];
    errno_t ret = memset_sp(handles, GS_MAX_DATA_FILES * sizeof(int32), -1, GS_MAX_DATA_FILES * sizeof(int32));
    knl_securec_check(ret);
    build_analyse_item_t *item = NULL;
    bak_t *bak = &session->kernel->backup_ctx.bak;
    char *buf = session->kernel->attr.xpurpose_buf;
    ret = memset_sp(buf, DEFAULT_PAGE_SIZE(session), 0, DEFAULT_PAGE_SIZE(session));
    knl_securec_check(ret);

    for (uint32 i = 0; i < bak->page_count; i++) {
        item = &bak->build_aly_items[i];
        if (item->page_id->file == GS_INVALID_FILEID) {
            continue;
        }

        datafile_t *df = DATAFILE_GET(session, item->page_id->file);
        if (cm_open_device(df->ctrl->name, df->ctrl->type, knl_io_flag(session), &handles[item->page_id->file]) != GS_SUCCESS) {
            GS_LOG_RUN_ERR("[REPAIR] failed to open datafile %s", df->ctrl->name);
            return GS_ERROR;
        }
        int64 offset = (int64)item->page_id->page * DEFAULT_PAGE_SIZE(session);
        if (cm_write_device(df->ctrl->type, handles[item->page_id->file], offset, buf, DEFAULT_PAGE_SIZE(session)) != GS_SUCCESS) {
            GS_LOG_RUN_ERR("[REPAIR] failed to write datafile %s", df->ctrl->name);
            bak_brain_close_file(handles, GS_MAX_DATA_FILES);
            return GS_ERROR;
        }
        GS_LOG_DEBUG_INF("[REPAIR] brain reset page %u-%u", item->page_id->file, item->page_id->page);
        item->page_id->file = GS_INVALID_FILEID;
    }

    for (uint32 i = 0; i < GS_MAX_DATA_FILES; i++) {
        if (handles[i] == GS_INVALID_HANDLE) {
            continue;
        }
        if (db_fdatasync_file(session, handles[i]) != GS_SUCCESS) {
            bak_brain_close_file(handles, GS_MAX_DATA_FILES);
            GS_LOG_RUN_ERR("[REPAIR] failed to fdatasync datafile");
            return GS_ERROR;
        }
    }

    bak_brain_close_file(handles, GS_MAX_DATA_FILES);
    return GS_SUCCESS;
}

static status_t bak_build_from_remote(knl_session_t *session)
{
    bak_context_t *ctx = &session->kernel->backup_ctx;
    bak_t *bak = &ctx->bak;
    bak_process_t *proc = &ctx->process[BAK_COMMON_PROC];

    if (bak_check_keep_alive(session) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (bak_build_start(session) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (bak_agent_send_build_ctrl(session, bak, proc) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (bak->record.is_repair && bak_agent_send_diff_pages(session) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (bak->record.attr.level > 0) {
        if (rst_build_prepare_basefiles(session, &ctx->process[BAK_COMMON_PROC]) != GS_SUCCESS) {
            return GS_ERROR;
        }
    }

    if (rst_build_restore(session) != GS_SUCCESS) {
        if (!bak->need_retry) {
            GS_LOG_RUN_INF("[RESTORE] restore failed, no more retry");
            (void)bak_agent_send_pkg(bak, BAK_PKG_ACK);
        }
        return GS_ERROR;
    }

    if (bak_agent_send_pkg(bak, BAK_PKG_ACK) != GS_SUCCESS) {
        return GS_ERROR;
    }

    rst_wait_write_end(bak);

    if (bak->record.is_repair) {
        if (bak_brain_reset_page(session) != GS_SUCCESS) {
            return GS_ERROR;
        }
    }

    return bak->failed ? GS_ERROR : GS_SUCCESS;
}

static status_t rst_build_clean_arch(knl_session_t *session)
{
    uint32 arch_num = 0;
    uint32 clean_locator = 0;
    arch_ctrl_t *arch_ctrl = NULL;
    arch_proc_context_t *proc_ctx = &session->kernel->arch_ctx.arch_proc[0];
    dtc_node_ctrl_t *node = dtc_my_ctrl(session);

    arch_get_files_num(session, ARCH_DEFAULT_DEST, session->kernel->id, &arch_num);

    for (uint32 i = 0; i < arch_num; i++) {
        clean_locator = (node->archived_start + i) % GS_MAX_ARCH_NUM;
        arch_ctrl = db_get_arch_ctrl(session, clean_locator, session->kernel->id);
        if (arch_ctrl->recid == 0) {
            continue;
        }

        if (!cm_exist_device(cm_device_type(arch_ctrl->name), arch_ctrl->name)) {
            GS_LOG_RUN_INF("archive file %s does not exist", arch_ctrl->name);
        } else {
            if (cm_remove_device(cm_device_type(arch_ctrl->name), arch_ctrl->name) != GS_SUCCESS) {
                GS_LOG_RUN_ERR("Failed to remove archive file %s", arch_ctrl->name);
                return GS_ERROR;
            }
        }

        arch_ctrl->recid = 0;
    }
    proc_ctx->curr_arch_size = 0;
    return GS_SUCCESS;
}

status_t bak_build_prepare_attr(knl_session_t *session, build_param_ctrl_t *ctrl)
{
    bak_t *bak = &session->kernel->backup_ctx.bak;
    bak_attr_t *attr = &bak->record.attr;

    attr->compress = ctrl->compress;
    attr->level = (ctrl->is_increment || ctrl->is_repair) ? GS_TRUE : GS_FALSE;
    bak->compress_ctx.compress_level = ctrl->compress_level;
    bak->record.is_repair = ctrl->is_repair;
    bak->record.is_increment = ctrl->is_increment;
    bak->proc_count = (ctrl->parallelism == 0) ? BUILD_DEFAULT_PARALLELISM : ctrl->parallelism;
    bak->backup_buf_size = ctrl->buffer_size;

    if (ctrl->base_lsn != GS_INVALID_LSN) {
        attr->base_lsn = ctrl->base_lsn;
        return GS_SUCCESS;
    }

    if (ctrl->is_increment) {
        if (rst_build_get_base_lsn(session, &ctrl->base_lsn) != GS_SUCCESS) {
            return GS_ERROR;
        }

        if (rst_build_clean_arch(session) != GS_SUCCESS) {
            return GS_ERROR;
        }
    } else if (ctrl->is_repair) {
        if (build_analyze_mem_init(session) != GS_SUCCESS) {
            return GS_ERROR;
        }

        if (build_repair_prepare(session, &ctrl->base_lsn) != GS_SUCCESS) {
            build_analyze_mem_free(session);
            return GS_ERROR;
        }

        if (rst_build_clean_arch(session) != GS_SUCCESS) {
            build_analyze_mem_free(session);
            return GS_ERROR;
        }
    }
    attr->base_lsn = ctrl->base_lsn;

    return GS_SUCCESS;
}

status_t bak_build_restore(knl_session_t *session, build_param_ctrl_t *ctrl)
{
    bak_context_t *ctx = &session->kernel->backup_ctx;
    lrcv_context_t *lrcv_ctx = &session->kernel->lrcv_ctx;
    bak_t *bak = &ctx->bak;

    GS_LOG_RUN_INF("[BUILD] build start, compress method %s, compress level %u, is increment %u, is repair %u, "
                   "buffer size %uM", bak_get_compress_name(ctrl->compress), ctrl->compress_level, ctrl->is_increment,
                   ctrl->is_repair, ctrl->buffer_size / SIZE_M(1));
    bak->need_retry = GS_FALSE;

    if (lrcv_ctx->thread.closed) {
        GS_THROW_ERROR(ERR_LRCV_NOT_READY);
        return GS_ERROR;
    }

    if (bak_set_running(session, ctx) != GS_SUCCESS) {
        GS_THROW_ERROR(ERR_BACKUP_IN_PROGRESS, "restore");
        return GS_ERROR;
    }

    if (bak_build_prepare_attr(session, ctrl) != GS_SUCCESS) {
        bak_unset_running(session, ctx);
        return GS_ERROR;
    }

    status_t status = bak_build_from_remote(session);
    if (status != GS_SUCCESS) {
        bak->failed = GS_TRUE;
    }

    if (!bak->need_retry) {
        GS_LOG_RUN_INF("[BUILD] backup remote finished, failed :%d", bak->failed);
    }

    if (bak->error_info.err_code != ERR_BACKUP_IN_PROGRESS && !bak->need_retry) {
        build_analyze_mem_free(session);
    }

    bak_build_end(session);
    return status;
}

static void bak_send_error(bak_t *bak, cs_pipe_t *pipe, int32 max_pkg_size)
{
    bak_agent_head_t head;
    int32 err_code;
    const char *error_msg = NULL;
    int32 receive_size;

    if (!bak->build_stopped && bak->need_retry) {
        return;
    }

    head.ver = BAK_AGENT_PROTOCOL;
    head.cmd = BAK_PKG_ERROR;

    if (bak->build_stopped) {
        err_code = ERR_BUILD_CANCELLED;
        error_msg = "Build has been cancelled";
    } else {
        cm_get_error(&err_code, &error_msg, NULL);
    }

    head.len = (uint32)(sizeof(bak_agent_head_t) + sizeof(int32) + strlen(error_msg) + 1);
    GS_LOG_RUN_WAR("[BACKUP] build failed, error:%d, %s", err_code, error_msg);

    if (cs_write_stream(pipe, (char *)&head, sizeof(bak_agent_head_t), max_pkg_size) != GS_SUCCESS) {
        return;
    }

    if (cs_write_stream(pipe, (char *)&err_code, sizeof(int32), max_pkg_size) != GS_SUCCESS) {
        return;
    }

    if (cs_write_stream(pipe, (char *)error_msg, (uint32)strlen(error_msg) + 1, max_pkg_size) != GS_SUCCESS) {
        return;
    }

    (void)cs_read_stream(pipe, (char *)&head, GS_NETWORK_IO_TIMEOUT, sizeof(bak_agent_head_t), (int32 *)&receive_size);
}

static status_t bak_build_wait_start(knl_session_t *session, cs_pipe_t *pipe, build_param_ctrl_t *ctrl,
    build_progress_t *build_progress)
{
    bak_agent_head_t head;
    int32 receive_size;

    if (cs_read_stream(pipe, (char *)&head, GS_NETWORK_IO_TIMEOUT, sizeof(bak_agent_head_t),
                       (int32 *)&receive_size) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (receive_size != sizeof(bak_agent_head_t)) {
        GS_THROW_ERROR(ERR_TCP_TIMEOUT_REMAIN, (int32)(sizeof(bak_agent_head_t) - receive_size));
        return GS_ERROR;
    }

    if (head.cmd != BAK_PKG_START) {
        GS_THROW_ERROR(ERR_NOT_EXPECTED_BACKUP_PACKET, BAK_PKG_START, head.cmd);
        return GS_ERROR;
    }

    if (cs_read_stream(pipe, (char *)ctrl, GS_NETWORK_IO_TIMEOUT, sizeof(build_param_ctrl_t),
                       (int32 *)&receive_size) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (receive_size != sizeof(build_param_ctrl_t)) {
        GS_THROW_ERROR(ERR_TCP_TIMEOUT_REMAIN, (int32)(sizeof(build_param_ctrl_t) - receive_size));
        return GS_ERROR;
    }

    if (cs_read_stream(pipe, (char *)build_progress, GS_NETWORK_IO_TIMEOUT, sizeof(build_progress_t),
        (int32 *)&receive_size) != GS_SUCCESS) {
        return GS_ERROR;
    }
    GS_LOG_RUN_INF("[BUILD] received standby build progress [%u/%u/%llu/%u/%u/%u/%d]", build_progress->stage,
        build_progress->file_id, build_progress->data_offset, build_progress->asn, build_progress->curr_file_index,
        build_progress->last_file_index, build_progress->start_time);

    if (receive_size != sizeof(build_progress_t)) {
        GS_THROW_ERROR(ERR_TCP_TIMEOUT_REMAIN, (int32)(sizeof(build_progress_t) - receive_size));
        return GS_ERROR;
    }

    return GS_SUCCESS;
}

static status_t bak_set_break_point_params(knl_session_t *session, bak_context_t *ctx, build_progress_t *build_progress)
{
    build_progress_t *local_build_progress = &ctx->bak.progress.build_progress;
    int32 temp_time_start;
    errno_t ret;

    switch (build_progress->stage) {
        case BACKUP_START:
            ctx->bak.is_first_link = GS_TRUE;
            ctx->bak.need_check = GS_FALSE;
            ret = memset_sp(local_build_progress, sizeof(build_progress_t), 0, sizeof(build_progress_t));
            knl_securec_check(ret);
            local_build_progress->start_time = (int32)session->kernel->scn;
            break;
        case BACKUP_DATA_STAGE:
        case BACKUP_LOG_STAGE:
        case BACKUP_HEAD_STAGE:
            ctx->bak.is_first_link = GS_FALSE;
            ctx->bak.need_check = GS_TRUE;
            ctx->bak.send_buf.buf_size = 0;
            temp_time_start = local_build_progress->start_time;
            *local_build_progress = *build_progress;
            local_build_progress->start_time = temp_time_start;
            ctx->bak.file_count = build_progress->curr_file_index;
            ctx->bak.curr_file_index = build_progress->curr_file_index;
            ctx->bak.progress.stage = build_progress->stage;
            break;
        default:
            GS_LOG_RUN_ERR("[BUILD] primary DB received unexpected build process stage, %u",
                ctx->bak.progress.build_progress.stage);
            return GS_ERROR;
    }
    GS_LOG_RUN_INF("[BUILD] received standby build progress [%u/%u/%llu/%u/%u/%u/%d]", build_progress->stage,
        build_progress->file_id, build_progress->data_offset, build_progress->asn,
        build_progress->curr_file_index, build_progress->last_file_index, build_progress->start_time);

    return GS_SUCCESS;
}

static status_t bak_get_page_list(knl_session_t *session)
{
    bak_t *bak = &session->kernel->backup_ctx.bak;
    uint32 read_size = 0;
    bool32 read_end = GS_FALSE;
    bak->remote.remain_data_size = 0;

    GS_LOG_RUN_INF("[BUILD] start to get page list from standby");
    if (rst_agent_read_head(bak, BAK_PKG_DATA, &read_size, &read_end) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (read_size > BUILD_ALY_MAX_PAGE_SIZE) {
        GS_LOG_RUN_ERR("[BUILD] get page list from standby failed, received size: %u, max size: %llu", read_size,
            BUILD_ALY_MAX_PAGE_SIZE);
        return GS_ERROR;
    }

    if (bak_agent_recv(bak, (char *)bak->build_aly_pages, (int32)read_size) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (bak_agent_send_pkg(bak, BAK_PKG_ACK) != GS_SUCCESS) {
        return GS_ERROR;
    }

    bak->page_count = read_size / sizeof(page_id_t);
    build_analyze_construct_hash(session);
    GS_LOG_RUN_INF("[BUILD] get page list from standby successfully, page count: %u", bak->page_count);
    return GS_SUCCESS;
}

status_t bak_build_backup(knl_session_t *session, cs_pipe_t *pipe, cs_packet_t *send_pack, cs_packet_t *recv_pack)
{
    bak_context_t *ctx = &session->kernel->backup_ctx;
    build_param_ctrl_t ctrl;
    bak_t *bak = &ctx->bak;
    build_progress_t *local_build_progress = &ctx->bak.progress.build_progress;
    build_progress_t build_progress;

    if (bak_build_wait_start(session, pipe, &ctrl, &build_progress) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (bak_precheck(session) != GS_SUCCESS) {
        bak_send_error(bak, pipe, (int32)cm_atomic_get(&session->kernel->attr.repl_pkg_size));
        return GS_ERROR;
    }

    if (bak_set_build_running(session, ctx, &build_progress) != GS_SUCCESS) {
        bak_send_error(bak, pipe, (int32)cm_atomic_get(&session->kernel->attr.repl_pkg_size));
        return GS_ERROR;
    }

    bak_set_build_params(&ctx->bak, pipe, send_pack, recv_pack, &ctrl);

    if (bak_set_break_point_params(session, ctx, &build_progress) != GS_SUCCESS) {
        bak_unset_build_running(session, ctx);
        bak_send_error(bak, pipe, (int32)cm_atomic_get(&session->kernel->attr.repl_pkg_size));
        return GS_ERROR;
    }

    if (bak_agent_write(bak, (char *)&local_build_progress->start_time, sizeof(int32)) != GS_SUCCESS) {
        bak_unset_build_running(session, ctx);
        bak_send_error(bak, pipe, (int32)cm_atomic_get(&session->kernel->attr.repl_pkg_size));
        return GS_ERROR;
    }

    if (ctrl.is_repair) {
        if (build_analyze_mem_init(session) != GS_SUCCESS) {
            bak_unset_build_running(session, ctx);
            bak_send_error(bak, pipe, (int32)cm_atomic_get(&session->kernel->attr.repl_pkg_size));
            return GS_ERROR;
        }
        if (bak_get_page_list(session) != GS_SUCCESS) {
            bak_unset_build_running(session, ctx);
            build_analyze_mem_free(session);
            bak_send_error(bak, pipe, (int32)cm_atomic_get(&session->kernel->attr.repl_pkg_size));
            return GS_ERROR;
        }
    }

    status_t status = bak_backup_proc(session);
    if (ctrl.is_repair) {
        build_analyze_mem_free(session);
    }

    if (status != GS_SUCCESS) {
        bak_send_error(bak, pipe, (int32)cm_atomic_get(&session->kernel->attr.repl_pkg_size));
    }

    return status;
}

void bak_stream_read_prepare(knl_session_t *session, bak_process_t *process, datafile_t *datafile, uint32 sec_id)
{
    bak_t *bak = &session->kernel->backup_ctx.bak;
    bak_assignment_t *assign_ctrl = &process->assign_ctrl;
    bak_ctrl_t *ctrl = &process->ctrl;
    errno_t ret;

    ret = strcpy_sp(ctrl->name, GS_FILE_NAME_BUFFER_SIZE, datafile->ctrl->name);
    knl_securec_check(ret);
    ctrl->type = datafile->ctrl->type;
    ctrl->offset = DEFAULT_PAGE_SIZE(session);

    assign_ctrl->start = ctrl->offset;
    assign_ctrl->end = ctrl->offset;
    assign_ctrl->file_id = datafile->ctrl->id;
    assign_ctrl->sec_id = 0;

    bak_record_new_file(bak, BACKUP_DATA_FILE, assign_ctrl->file_id, sec_id, 0, GS_FALSE, 0, 0);
}

status_t bak_send_stream_data(knl_session_t *session, bak_t *bak, bak_assignment_t *assign_ctrl)
{
    bak_process_t *proc = &session->kernel->backup_ctx.process[BAK_COMMON_PROC];
    bak_stream_buf_t *stream = &bak->send_stream;
    date_t start;
    bool32 file_end = GS_FALSE;
    uint32 send_size;
    uint64 last_read_offset = 0;
    uint64 curr_read_offset = 0;
    char *send_buf = NULL;

    for (;;) {
        if (bak_check_session_status(session) != GS_SUCCESS) {
            return GS_ERROR;
        }

        if (!cm_spin_try_lock(&stream->lock)) {
            cm_spin_sleep();
            continue;
        }

        file_end = (stream->read_offset == assign_ctrl->file_size);
        if (stream->data_size[stream->wid] == 0 && !file_end) {
            // no data in working buffer, need retry
            cm_spin_unlock(&stream->lock);
            cm_spin_sleep();
            continue;
        }
        curr_read_offset = stream->read_offset;

        stream->wid = !stream->wid;
        stream->fid = !stream->fid;
        stream->data_size[stream->wid] = 0; // reset working buffer data size
        cm_spin_unlock(&stream->lock);

        send_size = stream->data_size[stream->fid];
        send_buf = stream->bufs[stream->fid].aligned_buf;
        if (send_size > 0) {
            start = g_timer()->now;
            if (bak_agent_write(bak, send_buf, send_size) != GS_SUCCESS) {
                return GS_ERROR;
            }
            proc->stat.write_time += (g_timer()->now - start);
            proc->stat.write_size += send_size;
        }

        bak_update_progress(bak, (curr_read_offset - last_read_offset));
        last_read_offset = curr_read_offset;
        if (file_end) {
            break;
        }
    }

    return GS_SUCCESS;
}

void bak_init_send_stream(bak_t *bak, uint32 start, uint64 filesize, uint32 file_id)
{
    bak_stream_buf_t *stream_buf = &bak->send_stream;
    bak_read_cursor_t *cursor = &bak->read_cursor;

    stream_buf->curr_block_id = 0;
    stream_buf->read_offset = start;
    stream_buf->bakfile_size = 0;
    stream_buf->wid = 0;
    stream_buf->fid = 1;
    stream_buf->data_size[0] = 0;
    stream_buf->data_size[1] = 0;

    cursor->block_id = 0;
    cursor->file_id = file_id;
    cursor->file_size = filesize;
    cursor->offset = start;
    cursor->read_size = 0;
}

status_t bak_stream_send_end(bak_t *bak, bak_stream_buf_t *stream_buf)
{
    bak->backup_size = stream_buf->bakfile_size;

    if (bak_agent_send_pkg(bak, BAK_PKG_FILE_END) != GS_SUCCESS) {
        return GS_ERROR;
    }

    if (bak_agent_wait_pkg(bak, BAK_PKG_ACK) != GS_SUCCESS) {
        return GS_ERROR;
    }

    knl_panic(bak->files[bak->curr_file_index].type != BACKUP_HEAD_FILE);
    knl_panic(bak->curr_file_index < BAK_MAX_FILE_NUM);
    bak->files[bak->curr_file_index].size = bak->backup_size;
    bak->files[bak->curr_file_index].sec_start = 0;
    bak->files[bak->curr_file_index].sec_end = 0;
    bak->curr_file_index++;

    return GS_SUCCESS;
}

void rst_assign_uds_restore_task(knl_session_t *session, bak_process_t *proc)
{
    bak_t *bak = &session->kernel->backup_ctx.bak;
    uint32 bak_index = bak->file_count - 1;
    bak_file_t *bak_file = &bak->files[bak_index];
    bak_assignment_t *assign_ctrl = &proc->assign_ctrl;

    knl_panic(proc->is_free);
    assign_ctrl->task = bak->is_building ? BAK_BUILD_RESTORE_TASK : BAK_STREAM_RESTORE_TASK;
    bak_file->sec_start = assign_ctrl->section_start;
    bak_file->sec_end = assign_ctrl->section_end;

    for (uint32 j = 0; j < GS_MAX_DATA_FILES; j++) {
        knl_panic(proc->datafiles[j] == GS_INVALID_HANDLE);
    }

    CM_MFENCE;
    proc->is_free = GS_FALSE;
}

void rst_assign_stream_restore_task(knl_session_t *session, bak_ctrl_t *init_ctrl)
{
    bak_context_t *backup_ctx = &session->kernel->backup_ctx;
    bak_t *bak = &backup_ctx->bak;
    bak_assignment_t *assign_ctrl = NULL;
    bak_process_t *proc = NULL;
    bak_ctrl_t *ctrl = NULL;
    uint32 bak_index = bak->curr_file_index;

    for (uint32 id = 0; id < bak->proc_count; id++) {
        proc = &backup_ctx->process[id + 1];
        knl_panic(proc->is_free);

        ctrl = &proc->ctrl;
        *ctrl = *init_ctrl;
        ctrl->handle = GS_INVALID_HANDLE;

        assign_ctrl = &proc->assign_ctrl;
        assign_ctrl->sec_id = 0;
        assign_ctrl->start = 0;
        assign_ctrl->end = 0;
        assign_ctrl->section_start = 0;
        assign_ctrl->section_end = 0;
        assign_ctrl->bak_index = bak_index;
        assign_ctrl->bak_file.handle = bak->local.handle;
        assign_ctrl->type = bak->files[bak_index].type;

        rst_assign_uds_restore_task(session, proc);
    }
}

void rst_init_recv_stream(bak_t *bak)
{
    rst_stream_buf_t *recv_stream = &bak->recv_stream;

    recv_stream->prev_block = 0;
    recv_stream->curr_block_offset = 0;
    recv_stream->wid = 0;
    recv_stream->fid = 1;
    recv_stream->usable_size[0] = 0;
    recv_stream->usable_size[1] = 0;
    recv_stream->recv_size[0] = 0;
    recv_stream->recv_size[1] = 0;
    recv_stream->curr_file_tail = 0;
    recv_stream->is_eof = GS_FALSE;
}

status_t rst_stream_read_prepare(knl_session_t *session, rst_stream_buf_t *recv_stream, log_file_head_t *head,
                                 bool32 *ignore_logfile)
{
    bak_t *bak = &session->kernel->backup_ctx.bak;
    bak_process_t *common_proc = &session->kernel->backup_ctx.process[BAK_COMMON_PROC];
    uint32 bak_index = bak->curr_file_index;
    uint32 file_id = bak->files[bak_index].id;
    bool32 ignore_data = GS_FALSE;

    if (bak->files[bak_index].type == BACKUP_DATA_FILE) {
        recv_stream->base_filesize = common_proc->datafile_size[file_id];  // for incremental restore, set base file
                                                                           // size
    } else {
        int32 read_size;
        bool32 log_end = GS_FALSE;

        if (rst_agent_read(bak, (char *)head, sizeof(log_file_head_t), &read_size, &log_end) != GS_SUCCESS) {
            return GS_ERROR;
        }

        if (read_size < sizeof(log_file_head_t)) {
            GS_LOG_RUN_ERR("[RESTOR] failed to read log file head, read size %d less than %u", read_size,
                           (uint32)sizeof(log_file_head_t));
            return GS_ERROR;
        }

        if (log_verify_head_checksum(session, head, "") != GS_SUCCESS) {
            return GS_ERROR;
        }

        uint32 fill_size = CM_CALC_ALIGN(sizeof(log_file_head_t), (uint32)head->block_size) - sizeof(log_file_head_t);
        if (rst_agent_read(bak, ((char *)head) + sizeof(log_file_head_t), fill_size, &read_size, &log_end) !=
            GS_SUCCESS) {
            return GS_ERROR;
        }

        if (read_size != fill_size) {
            GS_LOG_RUN_ERR("[RESTOR] failed to read log file head, read size %d less than %u", read_size, fill_size);
            return GS_ERROR;
        }
        common_proc->ctrl.handle = GS_INVALID_HANDLE;
        if (rst_set_logfile_ctrl(session, bak->curr_file_index, head, &common_proc->ctrl, &ignore_data) != GS_SUCCESS) {
            return GS_ERROR;
        }
        if (!ignore_data) {
            if (rst_write_data(session, &common_proc->ctrl, (const char *)head, fill_size + sizeof(log_file_head_t)) !=
                GS_SUCCESS) {
                cm_close_device(common_proc->ctrl.type, &common_proc->ctrl.handle);
                return GS_ERROR;
            }
        }
        cm_close_device(common_proc->ctrl.type, &common_proc->ctrl.handle);
        bak_update_progress(bak, (uint64)(fill_size + sizeof(log_file_head_t)));
    }

    *ignore_logfile = ignore_data;
    return GS_SUCCESS;
}

/*
 * recieve buffer tail may be incomplete block, need set uasble size
 */
status_t rst_stream_aligned_blocks(rst_stream_buf_t *stream, uint32 recv_size, bool32 ignore_data)
{
    knl_panic(recv_size <= stream->buf_size);
    if (ignore_data || recv_size == 0) {
        stream->usable_size[stream->fid] = 0;
        stream->recv_size[stream->fid] = 0;
        return GS_SUCCESS;
    }

    uint32 usable_size = 0;
    char *buf = stream->bufs[stream->fid].aligned_buf;
    bak_block_head_t *block = (bak_block_head_t *)(buf + usable_size);

    while (block->block_size + usable_size <= recv_size) {
        if (block->magic_num != LOG_MAGIC_NUMBER) {
            GS_LOG_RUN_ERR("[RESTOR] invalid received block with id %u, size is %u", block->block_id,
                           block->block_size);
            return GS_ERROR;
        }

        usable_size += block->block_size;
        if (sizeof(bak_block_head_t) + usable_size > recv_size) {
            break;
        }
        block = (bak_block_head_t *)(buf + usable_size);
    }

    if (usable_size < sizeof(bak_block_head_t)) {
        GS_LOG_RUN_ERR("[RESTOR] invalid received block with id %u, size is %u", block->block_id, block->block_size);
        return GS_ERROR;
    }
    knl_panic(usable_size <= recv_size);

    stream->usable_size[stream->fid] = usable_size;
    stream->recv_size[stream->fid] = recv_size;
    return GS_SUCCESS;
}

// previous recieve buffer tail may be incomplete block, need copy it to current buffer head
uint32 rst_fix_incomplete_block(rst_stream_buf_t *stream)
{
    char *curr_recv_buf = stream->bufs[stream->fid].aligned_buf;
    char *prev_buf_tail = stream->bufs[stream->wid].aligned_buf + stream->usable_size[stream->wid];
    uint32 tail_size = stream->recv_size[stream->wid] - stream->usable_size[stream->wid];
    errno_t ret;

    if (tail_size > 0) {
        ret = memmove_s(curr_recv_buf, stream->buf_size, prev_buf_tail, tail_size);
        knl_securec_check(ret);
    }

    return tail_size;
}

status_t rst_stream_read_internal(knl_session_t *session, char *buf, uint32 buf_size, int32 *recv_size,
                                  bool32 *file_end)
{
    bak_t *bak = &session->kernel->backup_ctx.bak;
    bak_process_t *proc = &session->kernel->backup_ctx.process[BAK_COMMON_PROC];
    date_t start = g_timer()->now;

    if (rst_agent_read(bak, buf, buf_size, recv_size, file_end) != GS_SUCCESS) {
        return GS_ERROR;
    }
    proc->stat.read_time += (g_timer()->now - start);
    proc->stat.read_size += (*recv_size);
    bak_update_progress(bak, (uint64)(*recv_size));

    return GS_SUCCESS;
}

status_t rst_recv_stream_data(knl_session_t *session, bool32 ignore_data)
{
    bak_t *bak = &session->kernel->backup_ctx.bak;
    rst_stream_buf_t *stream = &bak->recv_stream;
    char *recv_buf = NULL;
    bool32 file_end = GS_FALSE;
    uint32 last_tail_size;
    int32 recv_size = 0;

    for (;;) {
        if (bak_check_session_status(session) != GS_SUCCESS) {
            return GS_ERROR;
        }

        if (!cm_spin_try_lock(&stream->lock)) {
            cm_spin_sleep();
            continue;
        }

        // make sure the data of working buffer has been distributed
        if (stream->usable_size[stream->wid] > stream->curr_block_offset) {
            // some data of working buffer is not distributed, need wait
            cm_spin_unlock(&stream->lock);
            cm_spin_sleep();
            continue;
        }

        knl_panic(stream->usable_size[stream->wid] == stream->curr_block_offset);
        stream->wid = !stream->wid;
        stream->fid = !stream->fid;
        stream->curr_block_offset = 0;
        recv_buf = stream->bufs[stream->fid].aligned_buf;

        if (file_end) {
            knl_panic(stream->recv_size[stream->wid] == stream->usable_size[stream->wid]);
            stream->is_eof = GS_TRUE;
            cm_spin_unlock(&stream->lock);
            break;
        }
        cm_spin_unlock(&stream->lock);

        last_tail_size = rst_fix_incomplete_block(stream);
        if (rst_stream_read_internal(session, recv_buf + last_tail_size, stream->buf_size - last_tail_size, &recv_size,
                                     &file_end) != GS_SUCCESS) {
            return GS_ERROR;
        }
        GS_LOG_DEBUG_INF("[BACKUP] read size :%u, last block tail size %u", recv_size, last_tail_size);
        recv_size += last_tail_size;
        if (rst_stream_aligned_blocks(stream, (uint32)recv_size, ignore_data) != GS_SUCCESS) {
            return GS_ERROR;
        }
    }

    return GS_SUCCESS;
}
